from pyspark.context import SparkContext
from pyspark.streaming.context import StreamingContext

sc = SparkContext()

# 1、创建流处理的环境
# 微批处理：每隔一段时间处理一次，并不是一条一条处理
ssc = StreamingContext(sc, batchDuration=5)

# 设置checkpoint路径，用于保存计算的状态
ssc.checkpoint("../../data/checkpoint")

# 2、读取数据
# 在服务器中启动socket nc -lk 8888
# DStream:流处理编程模型，底层也是RDD
lines_ds = ssc.socketTextStream("master", 8888)

# 算子代码每隔5秒执行一次
words_ds = lines_ds.flatMap(lambda line: line.split(","))

kv_ds = words_ds.map(lambda word: (word, 1))


# state：状态：之前的计算结果，单词数量
# data： 当前批次数据
def update_fun(data, state):
    # 计算当前批次单词的数量
    curr_num = sum(data)

    # 获取之前单词的数量
    if state is not None:
        # 当前批次单词的数量，加上之前的单词数量，并返回
        return curr_num + state
    else:
        return curr_num

# updateStateByKey： 有状态算子，
# 每一次计算都计算前一次状态进行累加计算
kv_ds.updateStateByKey(update_fun).pprint()

# 启动spark streaming程序
ssc.start()
ssc.awaitTermination()
ssc.stop()
